package com.xiaoyilin.interceptor;

import org.apache.kafka.clients.producer.ProducerInterceptor;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;

import java.util.Map;

/**
 * 多个拦截器形成拦截器链
 * 执行顺序按照interceptor.classes配置顺序来
 */
public class ChainInterceptor implements ProducerInterceptor<String, String> {

    private volatile long sendSuccess = 0;
    private volatile long sendFailure = 0;

    @Override
    public ProducerRecord<String, String> onSend(ProducerRecord<String, String> record) {
        String modifiedValue = "ChainInterceptor-" + record.value();
        /**
         * record属性不要修改否则与预期效果出现偏差
         * 若修改需要准确判断
         *
         * 修改key影响分区计算和影响broker端日志压缩功能
         */
        return new ProducerRecord<>(record.topic(), record.partition(), record.timestamp(), record.key(), modifiedValue, record.headers());
    }

    @Override
    public void onAcknowledgement(RecordMetadata metadata, Exception exception) {
        if (exception == null) {
            sendSuccess++;
        } else {
            sendFailure++;
        }
    }

    @Override
    public void close() {
        double successRadio = (double) sendSuccess / (sendFailure + sendSuccess);
        System.out.println("发送成功率=" + String.format("%f", successRadio * 100) + "%");
    }

    @Override
    public void configure(Map<String, ?> configs) {

    }


}
